package gbench.common.matlib.rdd;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collector;
import java.util.stream.Stream;

import gbench.common.matlib.MatlibCanvas;
import gbench.common.matlib.MatlibCanvas.*;

/**
 * Pair Vec 对儿值 向量
 * 一个 PVec就像一个 DNA 序列，他是由碱基对儿 BasePair(bp)作为基础元素的序列结构。
 *
 * @param <K> key 对儿值的键类型
 * @param <V> value 对儿值的值类型
 */
public class PVec<K, V> extends TVec<BasePair<K, V>, PVec<K, V>> implements IAttrs {

    /**
     * 构造函数 <br>
     * 核心函数
     *
     * @param pairs 键值对儿集合
     */
    public PVec(final BasePair<K, V>[] pairs) {
        super(pairs);
    }
    
    /**
     * 构造函数
     *
     * @param pairs 键值对儿集合
     */
    public PVec(final Seq<BasePair<K, V>> pairs) {
        super(pairs.data());
    }

    /**
     * 构造函数
     *
     * @param pairs 键值对儿集合
     */
    @SuppressWarnings("unchecked")
    public PVec(final Collection<BasePair<K, V>> pairs) {
        super(pairs.toArray(BasePair[]::new));
    }

    /**
     * 构造函数
     *
     * @param pairs 键值对儿集合
     */
    @SuppressWarnings("unchecked")
    public PVec(final Stream<BasePair<K, V>> pairs) {
        super(pairs.toArray(BasePair[]::new));
    }

    /**
     * 根据对象数组构建示例
     *
     * @param tt BasePair 对儿值几何
     */
    @Override
    public PVec<K, V> build(final BasePair<K, V>[] tt) {
        return new PVec<>(tt);
    }

    /**
     * 衍生化出一个 同构对象:向量(数组)级别的变换
     *
     * @param <T>    目标类型的类型
     * @param <U>    目标类型的类型
     * @param mapper 转换到目标类型类的的  映射 方法 : [(t,u)]->[(k,v)]
     * @return 内部元素类型为[(t, u)]类型的 的 PVec
     */
    public <T,U> PVec<T,U> derive(final Function<BasePair<K,V>[], BasePair<T,U>[]> mapper) {
        
        return new PVec<>(mapper.apply(this.data));
    }

    /**
     * 术语取自 haskell 的 functor <br>
     * 使用mapper转换集合data数据，生成(t,u)类型PVec
     *
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,(k,v)->(t,u)
     * @return U类型的数据流
     */
    public <T, U> PVec<T, U> fmap(final BiFunction<Long, BasePair<K, V>, BasePair<T, U>> mapper) {
        final var n = this.data.length;
        @SuppressWarnings("unchecked")
        final var xx = (BasePair<T, U>[]) new BasePair[n];
        for (int i = 0; i < data.length; i++)
            xx[i] = mapper.apply((long) i, data[i]);
        return new PVec<>(xx);
    }

    /**
     * 术语取自 haskell 的 functor <br>
     * 使用mapper转换集合data数据，生成u类型XVec
     *
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,t->u
     * @return U类型的数据流
     */
    public <T, U> PVec<T, U> fmap(final Function<BasePair<K, V>, BasePair<T, U>> mapper) {
       return this.fmap((i,p)->mapper.apply(p));
    }

    /**
     * 交换V与K的位置 fmap(BasePair::swap) 简写
     *
     * @return 交换V与K的位置的新的向量
     */
    public PVec<V,K> fswap(){
        return this.fmap(BasePair::swap);
    }

    /**
     * 二元变换:可以改变t值与类型的shuffle (i,t)->weightable <br>
     * 先为数据指定权重，而后给予重排顺序
     *
     * @param <U>    结果的元素类型
     * @param mapper 元素变换器,把T 变换成一个带有权重的 T值 : <br>
     *               (i:索引序号从0开始,t:元素值) -> weightable&lt;u&gt;
     * @return XVec &lt;U&gt;
     */
    public <T,U> PVec<T,U> shuffleX(final BiFunction<Integer, BasePair<K,V>, Weightable<BasePair<T,U>>> mapper) {
        return new PVec<>(this.stream((i,p)->mapper.apply(i.intValue(),p)).sorted().map(Weightable::value));
    }

    /**
     * 按照key 进行分组
     *
     * @return PVec 结构[(k,[vv])]
     */
    public PVec<K, XVec<V>> groupByKey() {
        return this.groupBy(MatlibCanvas.IVPair::key).entrySet().stream()
                    .map(BasePair.bpf(k->k, v->v.stream().map(BasePair::value).collect(XVec.xveclc()))
                ).collect(Range.aaclc(PVec::new));
    }

    /**
     * 按照key 对value进行规约
     * @param  reducer 规约函数：[v]->u
     * @return PVec 结构[(k,u)]
     */
    public <U> PVec<K, U> reduceByKey(final Function<? super XVec<V>,U> reducer) {
        return this.groupBy(MatlibCanvas.IVPair::key).entrySet().stream()
                    .map(BasePair.bpf(k->k, v->v.stream().map(BasePair::value).collect(XVec.xveclc()))
                ).map(BasePair.bpf2(k->k, reducer::apply)).collect(Range.aaclc(PVec::new));
    }

    /**
     * 按照key 对value进行规约
     * @param  collector 规约函数：[v]->u
     * @return PVec 结构[(k,u)]
     */
    public <U> PVec<K, U> reduceByKey(final Collector<? super V,?,U> collector) {
        return this.groupBy(MatlibCanvas.IVPair::key).entrySet().stream()
                .map(BasePair.bpf(k->k, v->v.stream().map(BasePair::value).collect(collector))
                ).collect(Range.aaclc(PVec::new));
    }

    /**
     * 按照value 对k进行分组
     *
     * @return PVec 结构[(v,[k])]
     */
    public PVec<V, XVec<K>> groupByValue(){
        return this.fswap().groupByKey();
    }

    /**
     * 按照value 对k进行规约
     * @param  collector 规约函数：[k]->u
     * @return PVec 结构[(v,u)]
     */
    public <U> PVec<V, U> reduceByValue(final Collector<? super K,?,U> collector){
        return this.fswap().reduceByKey(collector);
    }
    
    /**
     * 使用mapper转换集合data数据，生成u类型的数据流
     * 
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,t->u
     * @return U类型XVec
     */
    public <U> XVec<U> mapX(final Function<BasePair<K, V>, U> mapper) {

        return this.map(mapper).collect(XVec.xveclc());
    }

    /**
     * 使用mapper转换集合data数据，生成u类型的数据流
     * 
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,(i:从0开始,t:数据元素)->u
     * @return U类型XVec
     */
    public <U> XVec<U> mapX(final BiFunction<Long, BasePair<K, V>, U> mapper) {

        return this.mapX(mapper).collect(XVec.xveclc());
    }
    
    /**
     * 使用mapper转换集合data数据，生成U类型XVec
     * 
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,t->[u]
     * @return U类型的数据流
     */
    public <U> XVec<U> flatMapX(final Function<BasePair<K, V>, Stream<U>> mapper) {
        
        return this.flatMap(mapper).collect(XVec.xveclc());
    }
    
    /**
     * 使用mapper转换集合data数据，生成U类型XVec
     * 
     * @param <U>    结果的元素类型
     * @param mapper 数据流映射函数,(i:从0开始,t:数据元素)->[u]
     * @return U类型XVec
     */
    public <U> XVec<U> flatMapX(final BiFunction<Long, BasePair<K, V>, Stream<U>> mapper) {

        return this.flatMap(mapper).collect(XVec.xveclc());
    }

    /**
     * 按照value 进行规约
     * @param  reducer 规约函数：[k]->u
     * @return PVec 结构[(v,u)]
     */
    public <U> PVec<V, U> reduceByValue(final Function<? super XVec<K>,U> reducer){
        return this.fswap().reduceByKey(reducer);
    }

    /**
     * 值变换结果
     * @param <T> 变换结果类型
     * @param mapper1 键名-一位元素变换器 v->t
     * @return PVec&lt;K,T&gt;
     */
    public <T> PVec<T,V> fmap1(final Function<K,T> mapper1){
        return this.fmap(BasePair.bpf2(mapper1, e->e));
    }

    /**
     * 值变换结果
     * @param <T> 变换结果类型
     * @param mapper2 值-二位元素变换器 v->t
     * @return PVec&lt;K,T&gt;
     */
    public <T> PVec<K,T> fmap2(final Function<V,T> mapper2){
        return this.fmap(BasePair.bpf2(e->e,mapper2));
    }

    /**
     * 返回数据形状:依据索引序列所排成各个索引维度长度的数字列表。 <br>
     * 数据形状按照car的元素进行估计核算，注意返回的shapehi估计大小， <br>
     * 即假设XVec元素是均质的,统计类型包括:数组,XVec,Collection三类。<br>
     *
     * @return 当前向量的形状[第一维度, 第二维度, ...]
     */
    public NVec shape() {
        return this.shape(NVec::new);
    }

    /**
     * BasePair bp<br>
     * 生成键值对儿:垂直分裂 转换成一个键值对儿,这个可以类比为 helicase DNA 解旋酶
     * @return 分成键值对儿 ([k],[v])
     */
    public BasePair<XVec<K>,XVec<V>> unwind(){
        final var kk = new ArrayList<K>(this.size());
        final var vv = new ArrayList<V>(this.size());
        for(final var p:this.data){
            kk.add(p.key());
            vv.add(p.value());
        }
        return BasePair.bp(XVec.OF(kk),XVec.OF(vv));
    }
    
    /**
     * BasePair bp <br>
     * 生成键值对儿:垂直分裂 转换成一个键值对儿,这个可以类比为 helicase DNA 解旋酶
     * 
     * @param <T> 第一位置结果类型
     * @param <U> 第二位置结果类型
     * @param collector0 第一位置归集器 [k]->t
     * @param collector1 第二位置归集器 [v]->u
     * @return 分成键值对儿 (t,u)
     */
    public <T,U> BasePair<T,U> unwind(final Collector<K,?,T> collector0, final Collector<V,?,U> collector1){
        final var kk = new ArrayList<K>(this.size());
        final var vv = new ArrayList<V>(this.size());
        for(final var p:this.data){
            kk.add(p.key());
            vv.add(p.value());
        }
        return BasePair.bp(kk.stream().collect(collector0),vv.stream().collect(collector1));
    }

    /**
     * BasePair Get Stream
     *
     * @param k 键名
     * @return 键名所标定的值[(值索引, 值对象)]
     */
    public Stream<BasePair<Long, V>> pgetS(final K k) {

        return this.map((i, p) -> {
            if (p == null) return null;
            if ((p.key() != null && p.key().equals(k)) || (p.key() == null && k == null)) {
                return BasePair.bp(i, p.value());
            } else {
                return null;
            }// if
        }).filter(Objects::nonNull);
    }

    /**
     * BasePair Get XVec
     *
     * @param k 键名
     * @return 键名所标定的值[(值索引, 值对象)]
     */
    public XVec<BasePair<Long, V>> pgetX(final K k) {

        return this.pgetS(k).collect(Range.aaclc(XVec::new));
    }

    /**
     * BasePair Get First
     *
     * @param k 键名
     * @return 键名所标定的值[(值索引, 值对象)]
     */
    public Optional<BasePair<Long, V>> pgetFirst(final K k) {

        return this.pgetS(k).findFirst();
    }

    /**
     * BasePair Get first One
     *
     * @param k 键名
     * @return 键名所标定的值[(值索引, 值对象)]
     */
    public BasePair<Long, V> pgetFirst2(final K k) {

        return this.pgetFirst(k).orElse(null);
    }

    /**
     * BasePair Get
     *
     * @param k 键名
     * @return 键名所标定的值[(值索引, 值对象)]
     */
    public V pget(final K k) {

        return this.pgetFirst(k).map(IVPair::value).orElse(null);
    }

    /**
     * BasePair Get
     *
     * @param <U>    变换结果类型
     * @param k      键名
     * @param mapper 值转换器 v->u
     * @return 值所转换成的结果U
     */
    public <U> U pget(final K k,final Function<V,U> mapper) {

        return this.pgetFirst(k).map(IVPair::value).map(mapper).orElse(null);
    }

    /**
     * BasePair Get
     *
     * @param <U>    变换结果类型
     * @param k      键名
     * @param mapper 值转换器 (i:值索引从0开始,v:值对象)->u
     * @return 值所转换成的结果U
     */
    public <U> U pget(final K k,final BiFunction<Long,V,U> mapper) {

        return this.pgetFirst(k).map(p->mapper.apply(p.key(),p.value())).orElse(null);
    }

    /**
     * 获取属性集合
     */
    @Override
    public Map<Object, Object> getAttributes() {
        if(this.attrs==null) {
            synchronized(this){this.attrs = new HashMap<>();};
        }
        return attrs;
    }

    /**
     * 拉链归并器
     *
     * @param <K> 第一数据类型
     * @param <V> 第二数据类型
     * @param vv  第二位置向量,当长度不足之后,会循环补填以实现数据对应。
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<K, ?, PVec<K,V>> zipclc(final V[] vv) {
        final var ai = new AtomicInteger();
        final var n = vv.length;
        return Collector.of(ArrayList::new,
                (List<BasePair<K, V>> aa, K k) -> aa.add(new BasePair<>(k,vv[ai.getAndIncrement() % n])),
                (aa, bb) -> { aa.addAll(bb); return aa;},
                PVec::new);
    }
    
    /**
     * 拉链归并器
     *
     * @param <K> 第一数据类型
     * @param <V> 第二数据类型
     * @param vv  第二位置向量,当长度不足之后,会循环补填以实现数据对应。
     * @param bpattrs bp的属性集合
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<K, ?, PVec<K, V>> zipclc(final V[] vv,final Map<? extends Object, ? extends Object> bpattrs) {
        
        final var ai = new AtomicInteger();
        final var n = vv.length;
        return Collector.of(ArrayList::new,
                (List<BasePair<K, V>> aa, K k) -> aa.add(new BasePair<>(k,vv[ai.getAndIncrement() % n],bpattrs)),
                (aa, bb) -> { aa.addAll(bb); return aa;},
                PVec::new);
    }

    /**
     * 拉链归并器
     *
     * @param <K>  第一数据类型
     * @param <V>  第二数据类型
     * @param list 第二位置向量,当长度不足之后,会循环补填以实现数据对应。
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<K, ?, PVec<K,V>> zipclc(final List<V> list) {
        final var vv = list.stream().collect(Range.aaclc(e->e));
        return zipclc(vv);
    }

    /**
     * 拉链归并器
     *
     * @param <K> 第一数据类型
     * @param <V> 第二数据类型
     * @param seq 合并向量,当长度不足之后,会循环补填以实现数据对应。
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<K, ?, PVec<K,V>> zipclc(final Seq<V> seq) {
        return zipclc(seq.data());
    }
    
    /**
     * 拉链归并器
     *
     * @param <K>     第一数据类型
     * @param <V>     第二数据类型
     * @param seq     合并向量,当长度不足之后,会循环补填以实现数据对应。
     * @param bpattrs bp的属性集合
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<K, ?, PVec<K, V>> zipclc(final Seq<V> seq,
            final Map<? extends Object, ? extends Object> bpattrs) {
        return zipclc(seq.data(), bpattrs);
    }

    /**
     * 单值 归集,pvec归集器,pvec1 表示这是一个 单值  归集。
     *
     * @param <K>        源数据类型
     * @param <V>        目标数据类型
     * @param classifier 分类器 v->k
     * @return 把T类型的流归结成PVec&lt;K,U&gt; 类型的归集器
     */
    public static <K, V> Collector<V, ?, PVec<K, V>> pvec1clc(final Function<V, K> classifier) {
        return Collector.of(ArrayList::new,
            (List<BasePair<K, V>> aa, V v) -> aa.add(new BasePair<>(classifier.apply(v), v)),
            (aa, bb) -> {aa.addAll(bb);return aa;},
            PVec::new);
    }

    /**
     * 单值 归集,pvec归集器，pvec1 表示这是一个 单值  归集。
     *
     * @param <K>        源数据类型
     * @param <V>        目标数据类型
     * @param classifier 分类器 (i:索引序号从0开始,v:元素数据)->k
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <K, V> Collector<V, ?, PVec<K, V>> pvec1clc(final BiFunction<Long,V, K> classifier) {
        final var al = new AtomicLong(0L);
        return Collector.of(ArrayList::new,
            (List<BasePair<K, V>> aa, V v) -> aa.add(new BasePair<>(classifier.apply(al.getAndIncrement(),v), v)),
            (aa, bb) -> {aa.addAll(bb);return aa;},
            PVec::new);
    }

    /**
     * 对儿值 归集,pvec归集器:pvec2 表示这是一个 对儿值  归集。注意，此处的 2不是 to的谐音
     *
     * @param mapper 元素变换器 (k,v)->(t,u)
     * @param <P>    IVPair 类型的数据
     * @param <T>    源数据 键类型
     * @param <U>    源数据 值类型
     * @param <K>    目标数据 键类型
     * @param <V>    目标数据 值类型
     * @return 把T类型的流归结成PVec&lt;T,U&gt; 类型的归集器
     */
    @SuppressWarnings("unchecked")
    public static <P extends IVPair<T, U, ?>, K, V, T, U> Collector<P, ?, PVec<K, V>>
    pvec2clc(final Function<BasePair<T,U>, BasePair<K,V>> mapper) {
        return Collector.of(ArrayList::new,
                (aa, a) -> aa.add(mapper.apply(a instanceof BasePair ? (BasePair<T, U>) a : new BasePair<>(a))),
                (aa, bb) -> {aa.addAll(bb);return aa;},
                aa-> new PVec<>((List<BasePair<K, V>>) (Object) aa));
    }

    /**
     * 对儿值 归集,pvec归集器：pvec2 表示这是一个 对儿值  归集。注意，此处的 2不是 to的谐音
     *
     * @param <P> IVPair 类型的数据
     * @param <K> 数据 键类型
     * @param <V> 数据 值类型
     * @return 把T类型的流归结成PVec&lt;T,U&gt; 类型的归集器
     */
    public static <P extends IVPair<K, V, ?>, K, V> Collector<P, ?, PVec<K, V>> pvec2clc() {
        return pvec2clc(e->e);
    }
    
    /**
     * 单值 归集,pvec归集器,pvec3 表示这是一个 对T类型(Three的开头字母,也是pvecXclc系列的发生序号)的元素 归集。
     *
     * @param <T>    源数据元素类型
     * @param <K>    键名类型
     * @param <V>    键值类型
     * @param mapper 元组转换器 t->(k,v)
     * @return 把T类型的流归结成PVec&lt;K,U&gt; 类型的归集器
     */
    public static <T, K, V> Collector<T, ?, PVec<K, V>> pvec3clc(final Function<T, BasePair<K, V>> mapper) {
        return Collector.of(ArrayList::new, (List<BasePair<K, V>> aa, T t) -> aa.add(mapper.apply(t)), (aa, bb) -> {
            aa.addAll(bb);
            return aa;
        }, PVec::new);
    }

    /**
     * 单值 归集,pvec归集器,pvec3 表示这是一个 对T类型(Three的开头字母,也是pvecXclc系列的发生序号)的元素 归集。
     *
     * @param <T>    源数据元素类型
     * @param <K>    键名类型
     * @param <V>    键值类型
     * @param mapper 元组转换器 (i:索引序号从0开始,t:元素数据)->(k,v)
     * @return 把T类型的流归结成PVec&lt;K,V&gt; 类型的归集器
     */
    public static <T, K, V> Collector<T, ?, PVec<K, V>> pvec3clc(final BiFunction<Long, T, BasePair<K, V>> mapper) {
        final var al = new AtomicLong(0L);
        return Collector.of(ArrayList::new,
                (List<BasePair<K, V>> aa, T t) -> aa.add(mapper.apply(al.getAndIncrement(), t)), (aa, bb) -> {
                    aa.addAll(bb);
                    return aa;
                }, PVec::new);
    }

    /**
     * 转换成一个 对儿值PVec结构：xt 表示 XVec&lt;T&gt;
     *
     * @param <T>    源数据类型
     * @param <K>    PVec 的key 类型
     * @param <V>    PVec 的值 类型
     * @param mapper 键值变换函数 (i:元素索引从0开始,v:元素值)->p
     * @return PVec 对儿值向量
     */
    public static <T,K,V> Function<XVec<T>,PVec<K,V>> xt2pvecf(final BiFunction<Long,T, BasePair<K,V>> mapper){
        return xvec->xvec.fmap(mapper).collect(Range.aaclc(PVec::new));
    }

    /**
     * 转换成一个 对儿值PVec结构:xkv 表示 XVec&lt;T&gt;
     *
     * @return PVec 对儿值向量
     */
    public  static <T> Function<XVec<T>,PVec<Long,T>> xt2pvec(){
        return xvec->xvec.fmap((BiFunction<Long, T, BasePair<Long, T>>) BasePair::new).collect(Range.aaclc(PVec::new));
    }
    
    /**
     * 转换成一个 对儿值PVec结构:xtv 表示 XVec&lt;K,V&gt;
     * @param <K> PVec 的key 类型
     * @param <V> PVec 的值 类型
     * @return PVec 对儿值向量
     */
    public static <K,V> Function<XVec<BasePair<K,V>>,PVec<K,V>> xkv2pvec(){
        return xvec->xvec.collect(Range.aaclc(PVec::new));
    }
    
    private Map<Object,Object> attrs; // 属性集合

}